package kafka

import (
	"context"
	"fmt"
	"github.com/Shopify/sarama"
	"sync"
)

var Address = []string{"10.16.48.44:9092"}

func ConsumerTest() {
	topic := []string{"test"}
	var wg = &sync.WaitGroup{}
	wg.Add(1)
	//广播式消费：消费者1
	go ClusterConsumer(wg, Address, topic, "group-1")
	//广播式消费：消费者2
	//go clusterConsumer(wg, Address, topic, "group-2")

	wg.Wait()
}

type exampleConsumerGroupHandler struct{}

func (exampleConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (exampleConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h exampleConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		fmt.Printf("Message topic:%q partition:%d offset:%d key:%s value:%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value)
		sess.MarkMessage(msg, "")
	}
	return nil
}

// 支持brokers cluster的消费者
func ClusterConsumer(wg *sync.WaitGroup, brokers, topics []string, groupId string) {
	defer wg.Done()
	config := sarama.NewConfig()
	config.Version = sarama.V2_5_0_0 // specify appropriate version
	config.Consumer.Return.Errors = true

	group, err := sarama.NewConsumerGroup(brokers, groupId, config)
	if err != nil {
		panic(err)
	}
	defer func() { _ = group.Close() }()

	// Track errors
	go func() {
		for err := range group.Errors() {
			fmt.Println("ERROR", err)
		}
	}()

	// Iterate over consumer sessions.
	ctx := context.Background()
	for {
		handler := exampleConsumerGroupHandler{}

		// `Consume` should be called inside an infinite loop, when a
		// server-side rebalance happens, the consumer session will need to be
		// recreated to get the new claims
		err := group.Consume(ctx, topics, handler)
		if err != nil {
			panic(err)
		}
	}
}
